package com.gopay.common.logging;

import java.util.List;

import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
import redis.clients.jedis.exceptions.JedisConnectionException;

import com.gopay.common.util.RedisUtil;

public class LogsConsumeJob implements Job {

	private static JedisPoolConfig jedisPoolConfig;
	private static JedisPool jedisPool;
	private static RedisUtil redisUtil;

	private static volatile boolean initialized = false;
	
	private static int maxAllowedErrorCount = 0;
	private static volatile int currentErrorCount = 0;
	private static String channel = "logsprod";

	@Override
	public void execute(JobExecutionContext context)
			throws JobExecutionException {

		if (!initialized) {
			String host = (String) context.getMergedJobDataMap().get("host");
			int port = (int) context.getMergedJobDataMap().get("port");

			int maxAllowedErrorCount = (int) context.getMergedJobDataMap().get("maxAllowedErrorCount");
			
			int jedisPoolMaxTotal = (int) context.getMergedJobDataMap().get("jedisPoolMaxTotal");
			int jedisPoolMaxIdle = (int) context.getMergedJobDataMap().get("jedisPoolMaxIdle");
			int jedisPoolMinIdle = (int) context.getMergedJobDataMap().get("jedisPoolMinIdle");
			int jedisPooMaxWaitMillis = (int) context.getMergedJobDataMap().get("jedisPooMaxWaitMillis");
			String channel = (String) context.getMergedJobDataMap().get("channel");

			if (channel != null) {
				LogsConsumeJob.channel = channel;
			}
			
			LogsConsumeJob.maxAllowedErrorCount = maxAllowedErrorCount;

			jedisPoolConfig = new JedisPoolConfig();
			jedisPoolConfig.setMaxTotal(jedisPoolMaxTotal);
			jedisPoolConfig.setMaxIdle(jedisPoolMaxIdle);
			jedisPoolConfig.setMinIdle(jedisPoolMinIdle);
			jedisPoolConfig.setMaxWaitMillis(jedisPooMaxWaitMillis);
			jedisPoolConfig.setTestOnBorrow(true);
			jedisPoolConfig.setTestOnReturn(true);
			jedisPoolConfig.setTestWhileIdle(true);

			jedisPool = new JedisPool(jedisPoolConfig, host, port, 300);
			redisUtil = new RedisUtil();
			redisUtil.setJedisPool(jedisPool);
			
			initialized = true;
		}
		
		List<String> list = LogsCollector.getInstance().getAndClear();
		
		if (LogsConsumeJob.maxAllowedErrorCount == 0
				|| (LogsConsumeJob.maxAllowedErrorCount > LogsConsumeJob.currentErrorCount)) {
			Jedis jedis = null;
			boolean borrowOrOprSuccess = true;
			try {
				jedis = redisUtil.getConnection();

				for (String log : list) {
					jedis.publish(LogsConsumeJob.channel, log);
				}
				
				LogsConsumeJob.currentErrorCount = 0;
			} catch (JedisConnectionException e) {
				borrowOrOprSuccess = false;
				redisUtil.closeBrokenConnection(jedis);
				e.printStackTrace();
				LogsConsumeJob.currentErrorCount++;
			} catch (Throwable e) {
				e.printStackTrace();
				LogsConsumeJob.currentErrorCount++;
			} finally {
				if (borrowOrOprSuccess) {
					redisUtil.closeConnection(jedis);
				}
			}
		}
	}

}
